package org.jeecg.modules.gift.component;


import cn.hutool.core.util.HexUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttProperties;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttEndpoint;
import io.vertx.mqtt.MqttServer;
import io.vertx.mqtt.MqttServerOptions;
import io.vertx.mqtt.MqttTopicSubscription;
import io.vertx.mqtt.messages.codes.MqttSubAckReasonCode;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.dto.message.MessageDTO;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.system.api.ISysBaseAPI;
import org.jeecg.common.util.DateUtils;
import org.jeecg.common.util.oConvertUtils;
import org.jeecg.modules.gift.entity.*;
import org.jeecg.modules.gift.service.*;
import org.jeecg.modules.system.entity.SysUser;
import org.jeecg.modules.system.mapper.SysUserMapper;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;

@Component
@Slf4j
public class CardReaderServerRunner implements ApplicationRunner, Job {

    //存储每个topic的订阅关系
    private static Map<String, List<MqttEndpoint>> topicSubscribers;

    //存储每个设备订阅的topic
    private static Map<MqttEndpoint, List<String>> subscriptions;

    private static Map<String, Long> cardReaderMap = new HashMap<>();

    MqttServerOptions options;

    @Value("${spring.profiles.active}")
    private String env;

    @Autowired
    IGiftCardReaderService giftCardReaderService;

    @Autowired
    private IGiftActivityUserService giftActivityUserService;

    @Autowired
    private IGiftMerchantService giftMerchantService;

    @Autowired
    private IGiftActivityMerchantService giftActivityMerchantService;

    @Autowired
    private IGiftActivityClockinService giftActivityClockinService;

    @Autowired
    private IGiftActivityService giftActivityService;

    @Resource
    private SysUserMapper userMapper;

    @Autowired
    private ISysBaseAPI sysBaseApi;


    @Override
    public void run(ApplicationArguments args) throws Exception {


        //生产环境
        if ("prod".equals(env)) {
            LambdaQueryWrapper<GiftCardReader> wrapper = new LambdaQueryWrapper<>();
            wrapper.eq(GiftCardReader::getNetworkStatus, "1");
            GiftCardReader reader = new GiftCardReader();
            reader.setNetworkStatus("0");
            giftCardReaderService.update(reader, wrapper);
        }

        cardReaderMap.clear();
        options = new MqttServerOptions();

        Vertx vertx = Vertx.vertx();
        MqttServer mqttServer = MqttServer.create(vertx, options);

        topicSubscribers = new HashMap<>();
        subscriptions = new HashMap<>();

        mqttServer.endpointHandler(endpoint -> {
            //使用lambda表达式实现了endpointHandler方法，传入参数endpoint；
            log.info("MQTT client [{}] request to connect, clean session = {} ", endpoint.clientIdentifier(), endpoint.isCleanSession());

            if (endpoint.auth() == null) {
                log.error("endpoint auth is null, client = {} ", endpoint.clientIdentifier());
                endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
                return;
            }

            log.info("username = {}, password = {} ", endpoint.auth().getUsername(), endpoint.auth().getPassword());

            LambdaQueryWrapper<GiftCardReader> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(GiftCardReader::getClientId, endpoint.clientIdentifier());
            GiftCardReader cardReader = giftCardReaderService.getOne(queryWrapper, false);
            if (cardReader == null) {
                log.error("client = {} unregistered", endpoint.clientIdentifier());
                endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
                return;
            }

            if (!cardReader.getUserName().equals(endpoint.auth().getUsername()) || !cardReader.getPassword().equals(endpoint.auth().getPassword())) {
                log.error("client = {}, incorrect username={} or password={}", endpoint.clientIdentifier(), cardReader.getUserName(), cardReader.getPassword());
                endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
                return;
            }

            cardReaderMap.put(endpoint.clientIdentifier(), System.currentTimeMillis());

            cardReader.setNetworkStatus("1");
            giftCardReaderService.updateById(cardReader);

            // accept connection from the remote client
            endpoint.accept(false);

            SubscribeHandle(endpoint);
            UnSubscribeHandle(endpoint);
            ReceiveHandle(endpoint);
            DisConnectHandle(endpoint);

        });

        mqttServer.listen(mqtt -> {
            if (mqtt.succeeded()) {
                log.info("MQTT server is listening on port = {} ", mqtt.result().actualPort());
            } else {
                log.error("Error on starting the server");
                mqtt.cause().printStackTrace();
            }
        });

    }

    /*
     * 设备断开处理
     * */
    private void DisConnectHandle(MqttEndpoint endpoint) {
        endpoint.disconnectMessageHandler(disconnectMessage -> {
            log.info("Received disconnect from client, reason code = {} ", disconnectMessage.code());
            //对两个映射订阅关系的列表进行更新
            for (String topic : subscriptions.get(endpoint)) {
                topicSubscribers.get(topic).remove(endpoint);
                log.info("[{}]", topic);
            }
            subscriptions.remove(endpoint);

            LambdaQueryWrapper<GiftCardReader> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(GiftCardReader::getClientId, endpoint.clientIdentifier());
            GiftCardReader cardReader = giftCardReaderService.getOne(queryWrapper, false);
            if (cardReader != null) {
                cardReader.setNetworkStatus("0");
                giftCardReaderService.updateById(cardReader);
            }
        });
    }

    /*
     *处理订阅消息
     * */
    private void SubscribeHandle(MqttEndpoint endpoint) {

        endpoint.subscribeHandler(subscribe -> {
            Boolean IsValidTopic = false;

            //存储订阅消息中要订阅的topic的列表
            List<MqttTopicSubscription> topicSubscriptions = subscribe.topicSubscriptions();

            //存储订阅topic Qos级别的列表
            List<MqttSubAckReasonCode> reasonCodes = new ArrayList<>();


            //遍历列表
            for (MqttTopicSubscription s : topicSubscriptions) {
                //topic
                String topic = s.topicName();
                //Qos级别
                MqttQoS qos = s.qualityOfService();

                //判断topic是否合法
                if (!isValidTopic(topic)) {
                    //不合法则向设备发送消息

                    endpoint.publish(topic, Buffer.buffer("非法topic，topic不可包含空格"), qos, false, false);
                    continue;
                } else {
                    IsValidTopic = true;
                }
                log.info("Subscription for  {}  with QoS {}", topic, qos);

                reasonCodes.add(MqttSubAckReasonCode.qosGranted(qos));
                //判断是否已有此topic，如果有则直接添加，没有则新建键值对
                if (!topicSubscribers.containsKey(topic)) {
                    topicSubscribers.put(topic, new ArrayList<MqttEndpoint>());
                }
                topicSubscribers.get(topic).add(endpoint);
                //同上
                if (!subscriptions.containsKey(endpoint)) {
                    subscriptions.put(endpoint, new ArrayList<String>());
                }
                subscriptions.get(endpoint).add(topic);
            }

            if (IsValidTopic) {
                endpoint.subscribeAcknowledge(subscribe.messageId(), reasonCodes, MqttProperties.NO_PROPERTIES);
            }
        });

    }

    /*
     * 处理退订
     * */
    private void UnSubscribeHandle(MqttEndpoint endpoint) {
        endpoint.unsubscribeHandler(unsubscribe -> {
            //遍历要退订的topic
            for (String unsubscribedTopic : unsubscribe.topics()) {

//                topicSubscribers.get(unsubscribedTopic).remove(endpoint);
//
//                //如果某topic的订阅列表为空，删除topic
//                if (topicSubscribers.get(unsubscribedTopic).size() == 0) {
//                    topicSubscribers.remove(unsubscribedTopic);
//                }
//
//                subscriptions.get(endpoint).remove(unsubscribedTopic);
//
//                //同上
//                if (subscriptions.get(endpoint).size() == 0) {
//                    subscriptions.remove(endpoint);
//                }

                log.info("unsubscribed  {}  for {}", endpoint.auth().getUsername(), unsubscribedTopic);
            }
            endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
        });
    }

    private void ReceiveHandle(MqttEndpoint endpoint) {
        endpoint.publishHandler(publish -> {

            String topic = publish.topicName();
            Buffer payload = publish.payload();

            //对topic的合法性进行判断
            if (!isValidTopic(topic)) {
                endpoint.publish(topic, Buffer.buffer("非法topic，topic不可包含空格"), MqttQoS.AT_MOST_ONCE, false, false);
                return;
            }

            cardReaderMap.put(endpoint.clientIdentifier(), System.currentTimeMillis());

            //记录日志接收到设备发布的消息
            //log.info("Received message  [{}]  with QoS [{}]", publish.payload().toString(Charset.defaultCharset()), publish.qosLevel());
            byte[] content = publish.payload().getBytes();
            log.info("Received message  [{}]  with QoS [{}]", HexUtil.encodeHexStr(content), publish.qosLevel());

            if (publish.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
                endpoint.publishAcknowledge(publish.messageId());
            } else if (publish.qosLevel() == MqttQoS.EXACTLY_ONCE) {
                endpoint.publishReceived(publish.messageId());
            }

            //遍历订阅关系，进行消息发布
            for (Map.Entry<String, List<MqttEndpoint>> entry : topicSubscribers.entrySet()) {
                String subscribedTopic = entry.getKey();
                //被订阅的topic
                List<MqttEndpoint> subscribers = entry.getValue();
                //订阅上方topic的订阅者

                //判断消息发布的topic是否能和被设备订阅的topic按照规则匹配
                if (isTopicMatch(subscribedTopic, topic)) {
                    //若匹配，则遍历topic订阅者列表，并进行消息发布
                    for (MqttEndpoint subscriber : subscribers) {
                        subscriber.publish(topic, payload, publish.qosLevel(), publish.isDup(), publish.isRetain());
                    }
                }
            }

            if (content.length == 12 && content[0] == 0x02 && content[11] == 0x03) {
                String cardNo = HexUtil.encodeHexStr(Arrays.copyOfRange(content, 1, 5));

                LambdaQueryWrapper<GiftActivityUser> wrapper = new LambdaQueryWrapper<>();
                wrapper.eq(GiftActivityUser::getCardNo, cardNo);
                wrapper.in(GiftActivityUser::getJoinStatus, Arrays.asList("1", "2", "3"));

                GiftActivityUser activityUser = giftActivityUserService.getOne(wrapper, false);
                if (activityUser == null) {
                    log.error("cardNo = {}, card not bound to user", cardNo);
                    return;
                }

                GiftActivity activity = giftActivityService.getById(activityUser.getActivityId());
                if (activity == null) {
                    log.error("the activity does not exist");
                    return;
                }

                LambdaQueryWrapper<GiftCardReader> queryWrapper = new LambdaQueryWrapper<>();
                queryWrapper.eq(GiftCardReader::getClientId, endpoint.clientIdentifier());
                GiftCardReader cardReader = giftCardReaderService.getOne(queryWrapper, false);
                if (cardReader == null) {
                    log.error("card reader [{}] not bound", endpoint.clientIdentifier());
                    return;
                }
                LambdaQueryWrapper<GiftMerchant> mWrapper = new LambdaQueryWrapper<>();
                mWrapper.eq(GiftMerchant::getReaderId, cardReader.getId());

                GiftMerchant giftMerchant = giftMerchantService.getOne(mWrapper, false);
                if (giftMerchant == null) {
                    log.error("reader id = {}, merchant not found", cardReader.getId());
                    return;
                }

                LambdaQueryWrapper<GiftActivityMerchant> aWrapper = new LambdaQueryWrapper<>();
                aWrapper.eq(GiftActivityMerchant::getActivityId, activityUser.getActivityId());

                if (giftActivityMerchantService.count(aWrapper) == 0) {
                    log.error("商户[{}] 未参加活动ID=[{}]", giftMerchant.getMerchantName(), activityUser.getActivityId());
                    return;
                }

                LambdaQueryWrapper<GiftActivityClockin> cWrapper = new LambdaQueryWrapper<>();
                cWrapper.eq(GiftActivityClockin::getActivityId, activityUser.getActivityId());
                cWrapper.eq(GiftActivityClockin::getUserId, activityUser.getUserId());
                cWrapper.eq(GiftActivityClockin::getMailId, giftMerchant.getId());

                if (giftActivityClockinService.count(cWrapper) > 0) {
                    log.error("user [{}] already clock in, please do not clock in again", activityUser.getUserId());
                    return;
                }

                GiftActivityClockin clockin = new GiftActivityClockin();
                clockin.setActivityId(activityUser.getActivityId());
                clockin.setUserId(activityUser.getUserId());
                clockin.setMailId(giftMerchant.getId());
                giftActivityClockinService.save(clockin);

                SysUser user = userMapper.selectById(activityUser.getUserId());
                if (user != null)
                    activityCheckIn(user.getUsername(), activity.getActivityName(), giftMerchant.getMerchantName());
            }
        });
        endpoint.publishAcknowledgeHandler(messageId -> {
            log.info("received ack for message = {}", messageId);

        }).publishReceivedHandler(messageId -> {
            endpoint.publishRelease(messageId);
        }).publishCompletionHandler(messageId -> {
            log.info("received ack for message = {}", messageId);
        });
        endpoint.publishReleaseHandler(endpoint::publishComplete);
    }

    /*
     * 判断topic是否匹配
     * */
    private boolean isTopicMatch(String subscribedTopic, String publishedTopic) {
        String[] publishTopicArray = publishedTopic.split("/");
        String[] subscribedTopicArray = subscribedTopic.split("/");
        //将两个要比较的topic分割

        //订阅的topic长度不能比发布的topic长一个以上
        if (subscribedTopicArray.length - 1 > publishTopicArray.length) {
            return false;
        }

        //如果发布的topic长度比订阅的topic长度要长
        //并且订阅的topic最后不是以#结尾都返回false，因为这不可能
        if (subscribedTopicArray.length < publishTopicArray.length) {
            if (!subscribedTopicArray[subscribedTopicArray.length - 1].equals("#")) {
                return false;
            }
        }

        //对两个topic进行比较
        for (int i = 0; i < publishTopicArray.length && i < subscribedTopicArray.length; i++) {
            //如果匹配成功或者匹配到了+，进行下一层匹配
            if (subscribedTopicArray[i].equals(publishTopicArray[i]) || subscribedTopicArray[i].equals("+")) {
                continue;
            }

            //如果匹配到了#，直接通过
            if (subscribedTopicArray[i].equals("#")) {
                break;
            }
            return false;
        }
        return true;
    }

    public void activityCheckIn(String username, String activityName, String merchantName) {
        MessageDTO md = new MessageDTO();
        md.setToAll(false);
        md.setTitle("活动带卡信息");
        md.setTemplateCode("ACTIVITY_CHECK_IN");
        md.setToUser(username);
        md.setType("system");
        String testData = String.format("{\"activityName\":\"%s\",\"merchantName\":\"%s\"}", activityName, merchantName);
        if (oConvertUtils.isNotEmpty(testData)) {
            Map<String, Object> data = JSON.parseObject(testData, Map.class);
            md.setData(data);
        }
        sysBaseApi.sendTemplateMessage(md);
    }

    public boolean isValidTopic(String topic) {
        //topic 不能包含任何空格，并且要么以 /# 结尾，要么不包含 #
        return (!topic.matches(".*\\s+.*")) && (topic.matches(".*(?:\\/#)?$"));
    }

    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        log.info(String.format("检查读卡器是否在线,时间:" + DateUtils.getTimestamp()));
        cardReaderMap.keySet().forEach(key -> {
            long time = System.currentTimeMillis() - cardReaderMap.get(key);
            log.info("key={},time={}", key, time);
            if (time > 60 * 1000) {
                LambdaQueryWrapper<GiftCardReader> queryWrapper = new LambdaQueryWrapper<>();
                queryWrapper.eq(GiftCardReader::getClientId, key);
                GiftCardReader cardReader = giftCardReaderService.getOne(queryWrapper, false);
                if (cardReader != null) {
                    cardReader.setNetworkStatus("0");
                    giftCardReaderService.updateById(cardReader);
                }
                cardReaderMap.remove(key);
            }
        });
    }
}
